package com.dianzhang.disruptor.tools;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.Executors;

/**
 * @author xingyulai
 * @date 2024/2/21 2:35 PM
 * @description 创建内存队列工厂
 **/
@Component
public class DisruptorFactory {

    // 初始化RingBuffer的大小
    int ringBufferSize = 1024;

    @Autowired
    private DisruptorMessageHandler disruptorMessageHandler;

    private Disruptor<DisruptorMessage> disruptor;

    /**
     * 创建disrutor实例
     *
     * * @param eventHandler
     * @return
     */
    @PostConstruct
    private void init() {

        EventFactory<DisruptorMessage> factory = () -> {
            try {
                return DisruptorMessage.class.newInstance();
            } catch (InstantiationException e) {
                throw new RuntimeException(e);
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            }
        };
        //线程池会阻塞60s
        Disruptor<DisruptorMessage> disruptor = new Disruptor<>(factory, ringBufferSize, Executors.newCachedThreadPool());

//        disruptor.handleEventsWith(disruptorMessageHandler);

        disruptor.handleEventsWithWorkerPool(disruptorMessageHandler);


        disruptor.start();

        this.disruptor=disruptor;
    }

    @PreDestroy
    private  void destroy() {
        System.out.println("销毁了");
        disruptor.shutdown();
    }

    public void send(DisruptorMessage disruptorMessage){

        RingBuffer<DisruptorMessage> ringBuffer = disruptor.getRingBuffer();
        long sequence = ringBuffer.next();
        DisruptorMessage message = ringBuffer.get(sequence);
        message.setType(disruptorMessage.getType());
        message.setContent(disruptorMessage.getContent());
        ringBuffer.publish(sequence);
    }

    public <T> T getEventInstance(Disruptor<T> disruptor) {
        // 启动Disruptor
        disruptor.start();
        // 获取RingBuffer以便发布事件
        RingBuffer<T> ringBuffer = disruptor.getRingBuffer();

        long sequence = ringBuffer.next();

        T t = ringBuffer.get(sequence);

        return t;
    }





}
